package mq

import (
	"errors"
	"fmt"
	"sync"
)

type mqExit struct{}

type Mq struct {
	msg sync.Map
}

var myMq *Mq

func GetMq() *Mq {
	if myMq == nil {
		myMq = &Mq{}
	}
	return myMq
}

func (this *Mq) Create(topic string, buff chan interface{}) {
	_, ok := this.msg.Load(topic)
	if !ok {
		fmt.Println("add")
		this.msg.Store(topic, buff)
	}
}

func (this *Mq) Close(topic string) {
	vali, ok := this.msg.Load(topic)
	if ok {
		val := vali.(chan interface{})
		val <- mqExit{}
		this.msg.Delete(topic)
	}
}

func (this *Mq) Producer(topic string, buff interface{}) error {
	vali, ok := this.msg.Load(topic)
	if ok {
		val := vali.(chan interface{})
		val <- buff
	} else {
		return errors.New("no this topic")
	}
	return nil
}

func (this *Mq) Consumer(topic string, callback func(interface{})) error {
	vali, ok := this.msg.Load(topic)
	if ok {
		val := vali.(chan interface{})
		for {
			buff := <-val
			if buff == nil {
				continue
			}
			_, ok := buff.(mqExit)
			if ok {
				close(val)
				break
			}
			callback(buff)

		}
	} else {
		return errors.New("no this topic")
	}

	return nil
}
